package com.mq.mqserver.datacore;

import com.mq.common.MqException;
import com.mq.mqserver.core.Binding;
import com.mq.mqserver.core.Exchange;
import com.mq.mqserver.core.MSGQueue;
import com.mq.mqserver.core.Message;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 对内存中的数据进行管理:
 * 交换机  队列  绑定关系 消息  队列中的消息  队列中待确认的消息
 */
public class MemoryDataCenter {
//    交换机  key: exchangeName
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
//    队列: key:queueName
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
//   绑定关系 key1: exchangeName  key2: queueName
    private ConcurrentHashMap<String,ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
//  消息  key: messageId
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
//    队列中的消息数据:key1: queueName
    private ConcurrentHashMap<String,LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
//    队列中待确认的消息 key1: queueName  key2: messageId
    private ConcurrentHashMap<String,ConcurrentHashMap<String,Message>> queueMessageWritAckMap = new ConcurrentHashMap<>();

    //交换机操作
    //插入
    public void insertExchange(Exchange exchange){
        exchangeMap.put(exchange.getName(),exchange);
        System.out.println("[MemoryDataCenter] 新增交换机成功! exchangeName: "+exchange.getName());
    }
    //删除
    public void deleteExchange(String exchangeName){
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 交换机删除成功! exchangeName: "+exchangeName);
    }
    //查找
    public Exchange getExchange(String exchangeName){
        return exchangeMap.get(exchangeName);
    }

    //队列
    //插入
    public void insertQueue(MSGQueue queue){
        queueMap.put(queue.getName(),queue);
        System.out.println("[MemoryDataCenter] 新增队列成功! queueName: "+queue.getName());
    }
    //删除
    public void deleteQueue(String queueName){
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 队列删除成功! queueName: "+queueName);
    }
    //查找
    public MSGQueue getQueue(String queueName){
        return queueMap.get(queueName);
    }

    //绑定关系
    //插入
    public void insertBinding(Binding binding) throws MqException {
//        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
//        //先判断一下,该绑定关系是否已经存在,存在新值会将其覆盖,不存在就创建一个
//        if(bindingMap==null){
//            bindingMap = new ConcurrentHashMap<String,Binding>();
//        }
//        bindingMap.put(binding.getQueueName(),binding);
//        bindingsMap.put(binding.getExchangeName(),bindingMap);
        //对于键值对是否存在,ConcurrentHashMap有个更简洁的方法:先查询一下对应的哈希表是否存在,不存在就创建一个
        //computeIfAbsent(a,b):判断a是否存在,存在就赋值,不存在就执行b的拉姆达表达式,创建新的对象,和上面的逻辑是一样的
        //这个方法是原子操作:
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                f->new ConcurrentHashMap<>());
        //以交换机对应的哈希表为对象进行上锁,
        synchronized(bindingMap){
            Binding binding1 = bindingMap.get(binding.getQueueName());
            //当绑定关系已存在时,抛出异常,只有不存在时,才能插入
            if(binding1!=null){
                throw new MqException("[MemoryDataCenter] 绑定已存在 exchangeName: "+binding.getExchangeName()
                        +" ,queueName: "+binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(),binding);
//            bindingsMap.put(binding.getExchangeName(),bindingMap);
        }
        System.out.println("[MemoryDataCenter] 新的绑定创建成功! +exchangeName:"+binding.getExchangeName()
                +" ,queueName: "+binding.getQueueName()+" ,bindingKey: "+binding.getBindingKey());
    }
    //删除
    public void deleteBinding(Binding binding) throws MqException {
        //先判断该交换机是否存在绑定关系,不存在时,无法删除,抛出异常
        ConcurrentHashMap<String,Binding> bindingMap =  bindingsMap.get(binding.getExchangeName());
        if(bindingMap==null){
            throw new MqException("[MemoryDataCenter] 无绑定关系,删除失败 exchangeName: "+binding.getExchangeName()
                    +" ,queueName: "+binding.getQueueName());
        }
        //删除绑定
        bindingMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter] 绑定删除成功 exchangeName: "+binding.getExchangeName()
                +" ,queueName: "+binding.getQueueName()+" ,bindingKey:"+binding.getBindingKey());
    }
    //查找
    //根据交换机和队列查找指定绑定
    public Binding getBinding(String exchangeName,String queueName){
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if(bindingMap==null) {
            return null;
        }
        return bindingMap.get(queueName);
    }

    //查找交换机绑定的集合
    public ConcurrentHashMap<String,Binding> getBindings(String exchangeName){
        return bindingsMap.get(exchangeName);
    }
    //消息
    //添加
    public void insertMessage(Message message){
        messageMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 新消息添加成功 messageId: "+message.getMessageId());
    }
    //删除
    public void deleteMessage(String messageId){
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息删除陈功 messageId: "+messageId);
    }
    //查找
    public Message getMessage(String messageId){
        return messageMap.get(messageId);
    }

    //消息队列:
    //发送消息到指定的队列
    public void sendMessageToQueue(MSGQueue queue,Message message){
        //先查找该队列是否存在,不存在时,先创建:
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(),
                f -> new LinkedList<>());
        synchronized(messages){
            messages.add(message);
        }
        //将该消息在消息中心也存一份,就算消息中心已经存在该消息也没事,消息中心只会存一份相同的消息
        messageMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 消息发送到队列成功 queueName: "+queue.getName()+
                " ,messageId: "+message.getMessageId());
    }
    //从队列中取消息
    public Message pollMessageFromQueue(String queueName) throws MqException {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages==null){
            return null;
        }
        if(messages.isEmpty()){
            return null;
        }
        //当队列中有消息时,就从链表头部进行获取,并将其从链表中移除
        Message message = messages.remove(0);
        System.out.println("[MemoryDataCenter] 从消息队列中获取到消息 messageId: "+message.getMessageId());
        return message;
    }

    //获取指定队列中消息的个数
    public int getMessageCountFromQueue(String queueName) throws MqException {

        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages==null){
            //队列中没有消息
            return 0;
        }
        synchronized (messages){
            return messages.size();
        }
    }
    //未确认消息
    //添加未确认消息
    public void insertQueueMessageWaitAck(String queueName, Message message){
        ConcurrentHashMap<String, Message> messageMap = queueMessageWritAckMap.computeIfAbsent(queueName,
                f -> new ConcurrentHashMap<>());
        synchronized(messageMap){
            messageMap.put(message.getMessageId(),message);
        }
        System.out.println("[MemoryDataCenter] 新未确认消息添加到队列成功 messageId: "+message.getMessageId());
    }
    //删除未确认消息
    public void deleteWaitAckQueue(String queueName,String messageId){
        ConcurrentHashMap<String, Message> messageMap = queueMessageWritAckMap.get(queueName);
        if(messageMap==null){
            System.out.println("[MemoryDataCenter] 待确认队列中无消息,删除失败!");
        }
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 删除待确认消息成功!");
    }
    //获取指定的未确认消息
    public Message getWaitAckMessage(String queueName,String messageId){
        ConcurrentHashMap<String, Message> messageMap = queueMessageWritAckMap.get(queueName);
        if(messageMap==null){
            return null;
        }
        Message message = messageMap.get(messageId);
        if(message==null) {
            return null;
        }
        return message;
    }

    //恢复所有硬盘中的数据
    //当服务器重启后,内存中的所有数据都不存在了,需要从磁盘中恢复数据
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        //先将当前的集合都清空,防止残留数据造成混乱
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        queueMessageWritAckMap.clear();
        //恢复交换机数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();
        for(Exchange e:exchanges){
            String exchangeName = e.getName();
            exchangeMap.put(exchangeName,e);
        }
        //恢复队列数据
        List<MSGQueue> queues = diskDataCenter.selectAllQueues();
        for(MSGQueue q:queues){
            String queueName = q.getName();
            queueMap.put(queueName,q);
        }
        //恢复绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for(Binding binding:bindings){
            String exchangeName = binding.getExchangeName();
            String queueName = binding.getQueueName();
            //此时的绑定关系是可能不存在的,当不存在时,要进行创建:
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(exchangeName,
                    f->new ConcurrentHashMap<>());
            bindingMap.put(queueName,binding);
        }
        //恢复消息数据
        //查看队列中的消息,并将其加入到消息集合中
        for(MSGQueue q:queues){
            //获取队列中的所有消息
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(q.getName());
            for(Message message:messages){
                messageMap.put(message.getMessageId(),message);
            }
        }
        //对于未确认消息,当服务器重启后,服务器中所有的消息都要重新发送,未被确认的消息就都成了未被取走的消息了,
        //对于未确认的消息, 就不需要回复这些数据了
    }
}

